package com.xj.common.sdk.service.kafka;

import cn.hutool.core.convert.Convert;
import com.alibaba.fastjson.JSON;
import com.xj.common.bean.driver.DriverConfiguration;
import com.xj.common.bean.driver.DriverMetadata;
import com.xj.common.model.DriverEvent;
import com.xj.common.sdk.bean.driver.DriverContext;
import com.xj.common.sdk.service.DriverMetadataService;
import com.xj.common.sdk.service.DriverService;
import com.xj.common.utils.JsonPaseUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import com.xj.common.bean.constant.Common;

import javax.annotation.Resource;

/**
* 类说明 :接收KAFKA总线上驱动相关的处理信息，并做处理
*/
@Slf4j
@Component
public class DriverMetaConsumer {
    @Value("${spring.application.name}")
    private String serviceName;
    @Resource
    private DriverContext driverContext;
    @Resource
    private DriverService driverService;
    @Resource
    private DriverMetadataService driverMetadataService;
	 // 消费监听
    @KafkaListener(topics = {Common.KAFKA.TOPIC_EXCHANGE_METADATA},groupId = "metadataG2")
    public void onStatusMessage(ConsumerRecord<?, ?> record,Acknowledgment ack){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("DriverMetaConsumer简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
        String jsonstr = record.value().toString();
        DriverConfiguration driverConfiguration= JsonPaseUtil.JSONStrToObj(jsonstr, DriverConfiguration.class);
        //kafka没有单独通道，所以必须做服务识别
        // 不是本驱动的处理信息，不做处理
        if(!driverConfiguration.getServiceName().equals(serviceName))
            return ;
        switch (driverConfiguration.getType()) {
            case Common.Driver.Type.DRIVER:
                configurationDriver(driverConfiguration,ack);
                break;
            /*case Common.Driver.Type.PROFILE:
                configurationProfile(driverConfiguration);
                break;
            case Common.Driver.Type.DEVICE:
                configurationDevice(driverConfiguration);
                break;
            case Common.Driver.Type.POINT:
                configurationPoint(driverConfiguration);
                break;
            case Common.Driver.Type.DRIVER_INFO:
                configurationDriverInfo(driverConfiguration);
                break;
            case Common.Driver.Type.POINT_INFO:
                configurationPointInfo(driverConfiguration);
                break;*/
            default:
                break;
        }

        //提交偏移量
        ack.acknowledge();

    }
    /**
     * 配置 driver
     *
     * @param driverConfiguration DriverConfiguration
     */
    private void configurationDriver(DriverConfiguration driverConfiguration,Acknowledgment ack) {
        if (!Common.Response.OK.equals(driverConfiguration.getResponse())) {
            ack.acknowledge();
            driverService.close("The driver initialization failed {}", driverConfiguration.getResponse());
        }

        switch (driverConfiguration.getCommand()) {
            case Common.Driver.Event.DRIVER_HANDSHAKE_BACK:
                driverContext.setDriverStatus(Common.Driver.Status.REGISTERING);
                break;
            case Common.Driver.Event.DRIVER_REGISTER_BACK:
                driverContext.setDriverStatus(Common.Driver.Status.ONLINE);
                break;
            case Common.Driver.Event.DRIVER_METADATA_SYNC_BACK:
                DriverMetadata driverMetadata = Convert.convert(DriverMetadata.class, driverConfiguration.getContent());
                log.debug("Initialization driver metadata : {}", JSON.toJSONString(driverMetadata, true));
                driverContext.setDriverMetadata(driverMetadata);
                break;
            default:
                break;
        }
    }

}
